package com.laicunba.jms;

import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import com.laicunba.constant.GloabConstant;
import com.laicunba.util.StringUtil;
import com.laicunba.util.ThreadUtil;

/**
 * 
 * @author jiagoushi
 *
 */
@Component
public class JmsQueueListenerJob implements ApplicationListener<ContextRefreshedEvent> {
	private static final Logger LOG = Logger.getLogger(JmsQueueListenerJob.class);
	private static final String USER = GloabConstant.ACTIVEMQ_USERNAME;
	private static final String PASSWORD = GloabConstant.ACTIVEMQ_PASSWORD;
	private static final String URL = GloabConstant.ACTIVEMQ_URL;
	private static ActiveMQConnectionFactory FACTORY;
	private static Connection CONNECTION;
	private static boolean MOCK = StringUtil.isEmpty(URL);

	private static boolean canDurable = StringUtil.isNotEmpty(GloabConstant.ACTIVEMQ_CLIENTID);

	private static void initconn() {
		if (CONNECTION != null) {
			return;
		}
		ThreadUtil.execute(new Runnable() {

			@Override
			public void run() {
				try {
					MOCK = true;
					FACTORY = new ActiveMQConnectionFactory(URL);
					CONNECTION = FACTORY.createConnection(USER, PASSWORD);
					if (StringUtil.isNotEmpty(GloabConstant.ACTIVEMQ_CLIENTID)) {
						/**
						 * 因为需要持久化订阅，所以客户端ＩＤ不能随便变动
						 */
						CONNECTION.setClientID(GloabConstant.ACTIVEMQ_CLIENTID + ".listener." + InetAddress.getLocalHost().getHostName());
					}

					CONNECTION.start();
					MOCK = false;
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});

	}

	static {
		initconn();
	}

	private Map<String, JmsQueueListener> listeners = new HashMap<String, JmsQueueListener>();

	@Autowired
	private void setJmsQueueListener(JmsQueueListener[] listenerList) {
		for (JmsQueueListener listen : listenerList) {
			listeners.put("queue://" + listen.queue().toLowerCase(), listen);
		}
		LOG.info("JmsQueueListener : " + listeners);
	}

	private Map<String, JmsTopicListener> topicListeners = new HashMap<String, JmsTopicListener>();

	@Autowired
	private void setJmsTopicListener(JmsTopicListener[] listenerList) {
		if (listenerList == null) {
			return;
		}
		for (JmsTopicListener listen : listenerList) {
			topicListeners.put("topic://" + listen.topic().toLowerCase(), listen);
		}
		LOG.info("JmsTopicListener : " + topicListeners);
	}

	@Override
	public void onApplicationEvent(ContextRefreshedEvent event) {
		if (MOCK) {
			return;
		}
		if (listeners.isEmpty() && topicListeners.isEmpty()) {
			return;
		}

		try {
			Session session = CONNECTION.createSession(false, Session.AUTO_ACKNOWLEDGE);
			for (JmsQueueListener listen : listeners.values()) {
				MessageConsumer messageConsumer = session.createConsumer(new ActiveMQQueue(listen.queue()));
				messageConsumer.setMessageListener(new MessageListener() {

					@Override
					public void onMessage(Message message) {
						try {
							if (!listeners.containsKey(message.getJMSDestination().toString())) {
								return;
							}
							listeners.get(message.getJMSDestination().toString())._receive(message);
						} catch (Exception e) {
							LOG.error("", e);
						}

					}
				});
			}

			for (JmsTopicListener listen : topicListeners.values()) {
				MessageConsumer messageConsumer = null;
				if (canDurable && listen.isDurable()) {
					messageConsumer = session.createDurableSubscriber(new ActiveMQTopic(listen.topic()), listen.topic());
				} else {
					messageConsumer = session.createConsumer(new ActiveMQTopic(listen.topic()));
				}

				messageConsumer.setMessageListener(new MessageListener() {

					@Override
					public void onMessage(Message message) {
						try {

							if (!topicListeners.containsKey(message.getJMSDestination().toString())) {
								return;
							}
							topicListeners.get(message.getJMSDestination().toString()).receive(message);
						} catch (Exception e) {
							LOG.error("", e);
						}

					}
				});
			}

		} catch (JMSException e) {
			LOG.error("", e);
		}

	}

}
